/**
 * This file is part of Sonedyan.
 * 
 * Sonedyan is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public
 * License as published by the Free Software Foundation;
 * either version 3 of the License, or (at your option) any
 * later version.
 *
 * Sonedyan is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied
 * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
 * PURPOSE.  See the GNU General Public License for more
 * details.
 *
 * You should have received a copy of the GNU General Public
 * License along with Octave; see the file COPYING.  If not
 * see <http://www.gnu.org/licenses/>.
 * 
 * Copyright (C) 2009-2012 Jimmy Dubuisson <jimmy.dubuisson@gmail.com>
 */

package org.unige.mpej.eckmann.sonedyan.enron.db;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Date;
import java.util.Hashtable;
import java.util.List;
import java.util.Properties;
import java.util.Vector;

import javax.mail.Address;
import javax.mail.Message;
import javax.mail.internet.InternetAddress;
import javax.mail.Message.RecipientType;

import org.apache.log4j.Logger;
import org.unige.mpej.eckmann.sonedyan.enron.bean.Actor;
import org.unige.mpej.eckmann.sonedyan.enron.bean.Email;
import org.unige.mpej.eckmann.sonedyan.enron.bean.EnronMessage;
import org.unige.mpej.eckmann.sonedyan.enron.bean.Link;
import org.unige.mpej.eckmann.sonedyan.enron.bean.Actor.HierarchyLevel;
import org.unige.mpej.eckmann.sonedyan.enron.bean.Actor.Sex;
import org.unige.mpej.eckmann.sonedyan.enron.bean.Email.Type;

/**
 * Postgresql database handler
 */
public class PsqlHandler extends DatabaseHandler
{
	private static Logger log = Logger.getLogger(org.unige.mpej.eckmann.sonedyan.enron.db.PsqlHandler.class);
	
	private static final String HOST = "localhost";
	private static final String PORT = "5432";
	private static final String DB_NAME = "enron";
	private static final String USERNAME = "postgres";
	private static final String PASSWORD = "xxxxxxxx";
	private static final int DEFAULT_FETCH_SIZE = 1;
	
	private Connection conn;
	
	/**
	 * default constructor
	 */
	public PsqlHandler()
	{
		try
		{
			String url = "jdbc:postgresql://" + HOST + ":" + PORT + "/" + DB_NAME;
			Class.forName("org.postgresql.Driver");
			
			Properties props = new Properties();
			props.setProperty("user", USERNAME);
			props.setProperty("password", PASSWORD);

			this.conn = DriverManager.getConnection(url, props);
		}
		catch(Exception e)
		{
			log.error("An error occured: " + e.getMessage());
		}
	}
	
	/**
	 * execute specified select query
	 */
	public ResultSet execute(String query)
	{
		try
		{
			Statement s = this.conn.createStatement();
			return s.executeQuery(query);
		}
		catch(Exception e)
		{
			log.error("An error occured while executing query '" + query + "': " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * execute specified select prepared query
	 */
	public ResultSet execute(PreparedStatement s)
	{
		try
		{
			return s.executeQuery();
		}
		catch(Exception e)
		{
			log.error("An error occured while executing prepared query: " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * insert and get primary key
	 */
	public int executeAndGetKey(PreparedStatement s)
	{
		try
		{
			ResultSet result = this.execute(s);
			result.next();
			return result.getInt(1);
		}
		catch(Exception e)
		{
			log.error("An error occured while executing prepared query: " + e.getMessage());
			return -1;
		}
	}
	
	/**
	 * execute specified update query
	 */
	public int update(String query)
	{
		try
		{
			Statement s = this.conn.createStatement();
			return s.executeUpdate(query);
		}
		catch(Exception e)
		{
			log.error("An error occured while executing query '" + query + "': " + e.getMessage());
			return -1;
		}
	}
	
	/**
	 * execute specified update prepared query
	 */
	public int update(PreparedStatement s)
	{
		try
		{
			return s.executeUpdate();
		}
		catch(Exception e)
		{
			log.error("An error occured while executing prepared query: " + e.getMessage());
			return -1;
		}
	}
	
	/**
	 * close DB connection
	 */
	public boolean close()
	{
		try
		{
			this.conn.close();
			return true;
		}
		catch(Exception e)
		{
			log.error("Unable to close DB connection: " + e.getMessage());
			return false;
		}
	}
	
	/**
	 * store email
	 */
	public int storeEmail(String address)
	{
		try
		{
			Email email = new Email();
			email.setAddress(address);
			return this.storeEmail(email);
		}
		catch(Exception e)
		{
			log.error("Unable to store email: " + e.getMessage());
			return -1;
		}
	}
	
	/**
	 * store email
	 */
	public int storeEmail(Email email)
	{
		try
		{
			String address = email.getAddress().toString();
			Type type = email.getType();
			
			// try to retrieve email from the database
			Email storedEmail = this.getEmailByAddress(address);
			
			if (storedEmail != null)
			{
				if (storedEmail.getType() == null && type != null)
				{
					PreparedStatement s = this.conn.prepareStatement("UPDATE email SET type = ? WHERE address = ? RETURNING key");
					s.setString(1, type.toString());
					s.setString(2, address);
					
					return this.executeAndGetKey(s);
				}
				// nothing to do
				// return the email key
				// NB: could be more efficient
				else
				{
					PreparedStatement s = this.conn.prepareStatement("SELECT email.key FROM email WHERE address = ?");
					s.setString(1, address);
					
					ResultSet results = this.execute(s);
					results.next();
					
					return results.getInt("key");
				}
			}
			else
			{
				PreparedStatement s;
				
				if (type != null)
				{
					s = this.conn.prepareStatement("INSERT INTO email (address, type) VALUES (?, ?) RETURNING key");
					s.setString(1, address);
					s.setString(2, type.toString());
				}
				else
				{
					s = this.conn.prepareStatement("INSERT INTO email (address) VALUES (?) RETURNING key");
					s.setString(1, address);
				}
				
				return this.executeAndGetKey(s);
			}
		}
		catch(Exception e)
		{
			log.error("Unable to store email: " + e.getMessage());
			return -1;
		}
	}
	
	/**
	 * store link
	 */
	public int storeLink(Link link)
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("INSERT INTO link (from_actor_id, message_id, time, type, to_actor_id) VALUES (?, ?, ?, ?, ?)");
			s.setString(1, link.getFromActorId());
			s.setString(2, link.getLinkId());
			s.setDate(3, new java.sql.Date(link.getStartDate().getTime()));
			s.setString(4, link.getType().toString());
			s.setString(5, link.getToActorId());
			
			return this.update(s);
		}
		catch(Exception e)
		{
			log.error("Unable to store link: " + e.getMessage());
			return -1;
		}
	}
	
	/**
	 * get email
	 */
	public Email getEmailByAddress(String address)
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT type FROM email WHERE address = ?");
			s.setString(1, address);
			
			ResultSet results = this.execute(s);
			
			if (results.next())
			{
				Email email = new Email();
				String storedType = results.getString("type");
				email.setAddress(address);
				
				if (storedType != null)
				{
					if (storedType.equals("ENRON1")){ email.setType(Type.ENRON1); }
					else if (storedType.equals("ENRON2")){ email.setType(Type.ENRON2); }
					else if (storedType.equals("ENRON3")){ email.setType(Type.ENRON3); }
					else if (storedType.equals("NOT_ENRON_PRO")){ email.setType(Type.NOT_ENRON_PRO); }
					else if (storedType.equals("PRIVATE")){ email.setType(Type.PRIVATE); }
				}
				
				return email;
			}
			else
			{
				return null;
			}
		}
		catch(Exception e)
		{
			log.error("Unable to get email: " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * get email
	 */
	public Email getEmailByKey(int emailKey)
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT address, type FROM email WHERE key = ?");
			s.setInt(1, emailKey);
			
			ResultSet results = this.execute(s);
			
			if (results.next())
			{
				Email email = new Email();
				String storedAddress = results.getString("address");
				String storedType = results.getString("type");
				
				email.setAddress(storedAddress);
				
				if (storedType != null)
				{
					if (storedType.equals("ENRON1")){ email.setType(Type.ENRON1); }
					else if (storedType.equals("ENRON2")){ email.setType(Type.ENRON2); }
					else if (storedType.equals("ENRON3")){ email.setType(Type.ENRON3); }
					else if (storedType.equals("NOT_ENRON_PRO")){ email.setType(Type.NOT_ENRON_PRO); }
					else if (storedType.equals("PRIVATE")){ email.setType(Type.PRIVATE); }
				}
				
				return email;
			}
			else
			{
				return null;
			}
		}
		catch(Exception e)
		{
			log.error("Unable to get email: " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * add a join column entry
	 * 
	 * NB: an exception will be launched in case of duplicate entry
	 */
	private void addJoinEntry(String tableName, int key1, int key2)
	{
		try
		{
			PreparedStatement s;
			
			if (tableName.equals("actor_email"))
			{
				s = this.conn.prepareStatement("INSERT INTO actor_email (actor_key, email_key) VALUES (?, ?)");
				s.setInt(1, key1);
				s.setInt(2, key2);
			}
			else
			{
				s = this.conn.prepareStatement("INSERT INTO " + tableName + " (message_key, email_key) VALUES (?, ?)");
				s.setInt(1, key1);
				s.setInt(2, key2);
			}
			
			this.update(s);
		}
		catch(Exception e)
		{
			log.error("Unable to store join entry: " + e.getMessage());
		}
	}
	
	/**
	 * store message
	 */
	public int storeMessage(EnronMessage message)
	{
		try
		{
			// store sender email
			String senderAddress = message.getFrom().toString();
			int emailKey = this.storeEmail(senderAddress);
			
			PreparedStatement s = this.conn.prepareStatement("INSERT INTO message (id, sender_key, sent_date, subject, content, relative_path, number_of_to_recipients, number_of_cc_recipients, number_of_bcc_recipients) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING key");
			s.setString(1, message.getId());
			s.setInt(2, emailKey);
			s.setDate(3, new java.sql.Date(message.getSentDate().getTime()));
			s.setString(4, message.getSubject());
			s.setString(5, message.getMimeBodyAsString());
			s.setString(6, message.getRelativeFilePath());
			s.setInt(7, message.getNumberOfToRecipients());
			s.setInt(8, message.getNumberOfCcRecipients());
			s.setInt(9, message.getNumberOfBccRecipients());
			
			int messageKey = this.executeAndGetKey(s);
			
			// store TO, CC and BCC emails
			Address[] toAddresses = message.getToRecipients();
			
			if (toAddresses != null)
			{
				for (int i = 0; i < toAddresses.length; i++)
				{
					int key = this.storeEmail(toAddresses[i].toString());
					this.addJoinEntry("message_to", messageKey, key);
				}
			}
			
			Address[] ccAddresses = message.getCcRecipients();
			
			if (ccAddresses != null)
			{
				for (int i = 0; i < ccAddresses.length; i++)
				{
					int key = this.storeEmail(ccAddresses[i].toString());
					this.addJoinEntry("message_cc", messageKey, key);
				}
			}
			
			Address[] bccAddresses = message.getToRecipients();
			
			if (bccAddresses != null)
			{
				for (int i = 0; i < bccAddresses.length; i++)
				{
					int key = this.storeEmail(bccAddresses[i].toString());
					this.addJoinEntry("message_bcc", messageKey, key);
				}
			}
			
			return messageKey;
		}
		catch(Exception e)
		{
			log.error("Unable to store message: " + e.getMessage());
			return -1;
		}
	}
	
	/**
	 * store actor
	 */
	public int storeActor(Actor actor)
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("INSERT INTO actor (id, first_name, last_name, sex, hierarchy_level, job_description, is_known) VALUES (?, ?, ?, ?, ?, ?, ?) RETURNING key");
			s.setString(1, actor.getId());
			s.setBoolean(7, actor.isKnown());
			
			// potentially null values
			s.setString(2, actor.getFirstName());
			s.setString(3, actor.getLastName());
			
			Sex sex = actor.getSex();
			HierarchyLevel level = actor.getHierarchyLevel();
			
			if (sex != null){ s.setString(4, sex.toString()); } 
			else { s.setString(4, null); }
			if (level != null){ s.setString(5, level.toString());}
			else { s.setString(5, null); }
			
			s.setString(6, actor.getJobDescription());
			
			// insert actor
			int actorKey = this.executeAndGetKey(s);
			
			Vector<Email> emails = actor.getEmails();
			
			for (Email email : emails)
			{
				int emailKey = this.storeEmail(email);
				this.addJoinEntry("actor_email", actorKey, emailKey);
			}
			
			return actorKey;
		}
		catch(Exception e)
		{
			log.error("Unable to store actor: " + e.getMessage());
			e.printStackTrace();
			return -1;
		}
	}
	
	/**
	 * get actor by email
	 */
	public Actor getActorByAddress(String address)
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT actor.key, id, first_name, last_name, sex, hierarchy_level, job_description, is_known FROM actor LEFT JOIN actor_email ON actor.key = actor_email.actor_key LEFT JOIN email ON actor_email.email_key = email.key WHERE email.address = ?");
			s.setString(1, address);
			
			ResultSet results = this.execute(s);
			
			Actor actor = null;
			
			if (results != null && results.next())
			{
				actor = new Actor();
				
				int actorKey = results.getInt("key");
				
				actor.setId(results.getString("id"));
				actor.setFirstName(results.getString("first_name"));
				actor.setLastName(results.getString("last_name"));
				
				String sex = results.getString("sex");
				
				if (sex != null)
				{
					if (sex.equals("MALE")){ actor.setSex(Sex.MALE); }
					else { actor.setSex(Sex.FEMALE); }
				}
				
				String hierarchyLevel = results.getString("hierarchy_level");
				
				if (hierarchyLevel != null)
				{
					// CEO, PRESIDENT, VICE_PRESIDENT, MANAGING_DIRECTOR, DIRECTOR, MANAGER, LAWYER, TRADER, EMPLOYEE, UNKNOWN
					if (hierarchyLevel.equals("CEO")){ actor.setHierarchyLevel(HierarchyLevel.CEO); }
					else if (hierarchyLevel.equals("PRESIDENT")){ actor.setHierarchyLevel(HierarchyLevel.PRESIDENT); }
					else if (hierarchyLevel.equals("VICE_PRESIDENT")){ actor.setHierarchyLevel(HierarchyLevel.VICE_PRESIDENT); }
					else if (hierarchyLevel.equals("MANAGING_DIRECTOR")){ actor.setHierarchyLevel(HierarchyLevel.MANAGING_DIRECTOR); }
					else if (hierarchyLevel.equals("DIRECTOR")){ actor.setHierarchyLevel(HierarchyLevel.DIRECTOR); }
					else if (hierarchyLevel.equals("MANAGER")){ actor.setHierarchyLevel(HierarchyLevel.MANAGER); }
					else if (hierarchyLevel.equals("LAWYER")){ actor.setHierarchyLevel(HierarchyLevel.LAWYER); }
					else if (hierarchyLevel.equals("TRADER")){ actor.setHierarchyLevel(HierarchyLevel.TRADER); }
					else if (hierarchyLevel.equals("EMPLOYEE")){ actor.setHierarchyLevel(HierarchyLevel.EMPLOYEE); }
					else if (hierarchyLevel.equals("UNKNOWN")){ actor.setHierarchyLevel(HierarchyLevel.UNKNOWN); }
				}
				
				actor.setJobDescription(results.getString("job_description"));
				actor.setKnown(results.getBoolean("is_known"));
				
				// get the set of actor emails
				Vector<Email> emails = this.getActorEmails(actorKey);
				actor.setEmails(emails);
			}
			
			return actor;
		}
		catch(Exception e)
		{
			log.error("Unable to get actor by address ('" + address + "'): " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * get actor by email
	 */
	public Actor getActorById(String id)
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT actor.key, first_name, last_name, sex, hierarchy_level, job_description, is_known FROM actor WHERE id = ?");
			s.setString(1, id);
			
			ResultSet results = this.execute(s);
			
			Actor actor = null;
			
			if (results != null && results.next())
			{
				actor = new Actor();
				
				int actorKey = results.getInt("key");
				
				actor.setId(id);
				actor.setFirstName(results.getString("first_name"));
				actor.setLastName(results.getString("last_name"));
				
				String sex = results.getString("sex");
				
				if (sex != null)
				{
					if (sex.equals("MALE")){ actor.setSex(Sex.MALE); }
					else { actor.setSex(Sex.FEMALE); }
				}
				
				String hierarchyLevel = results.getString("hierarchy_level");
				
				if (hierarchyLevel != null)
				{
					// CEO, PRESIDENT, VICE_PRESIDENT, MANAGING_DIRECTOR, DIRECTOR, MANAGER, LAWYER, TRADER, EMPLOYEE, UNKNOWN
					if (hierarchyLevel.equals("CEO")){ actor.setHierarchyLevel(HierarchyLevel.CEO); }
					else if (hierarchyLevel.equals("PRESIDENT")){ actor.setHierarchyLevel(HierarchyLevel.PRESIDENT); }
					else if (hierarchyLevel.equals("VICE_PRESIDENT")){ actor.setHierarchyLevel(HierarchyLevel.VICE_PRESIDENT); }
					else if (hierarchyLevel.equals("MANAGING_DIRECTOR")){ actor.setHierarchyLevel(HierarchyLevel.MANAGING_DIRECTOR); }
					else if (hierarchyLevel.equals("DIRECTOR")){ actor.setHierarchyLevel(HierarchyLevel.DIRECTOR); }
					else if (hierarchyLevel.equals("MANAGER")){ actor.setHierarchyLevel(HierarchyLevel.MANAGER); }
					else if (hierarchyLevel.equals("LAWYER")){ actor.setHierarchyLevel(HierarchyLevel.LAWYER); }
					else if (hierarchyLevel.equals("TRADER")){ actor.setHierarchyLevel(HierarchyLevel.TRADER); }
					else if (hierarchyLevel.equals("EMPLOYEE")){ actor.setHierarchyLevel(HierarchyLevel.EMPLOYEE); }
					else if (hierarchyLevel.equals("UNKNOWN")){ actor.setHierarchyLevel(HierarchyLevel.UNKNOWN); }
				}
				
				actor.setJobDescription(results.getString("job_description"));
				actor.setKnown(results.getBoolean("is_known"));
				
				// get the set of actor emails
				Vector<Email> emails = this.getActorEmails(actorKey);
				actor.setEmails(emails);
			}
			
			return actor;
		}
		catch(Exception e)
		{
			log.error("Unable to get actor by id ('" + id + "'): " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * get actor emails
	 */
	public Vector<Email> getActorEmails(int actorKey)
	{
		try
		{
			Vector<Email> emails = new Vector<Email>();
			
			PreparedStatement s = this.conn.prepareStatement("SELECT address, type FROM email LEFT JOIN actor_email ON email.key = actor_email.email_key WHERE actor_email.actor_key = ?");
			s.setInt(1, actorKey);
			
			ResultSet results = this.execute(s);
			
			if (results != null)
			{
				while(results.next())
				{
					Email email = new Email();
					String storedAddress = results.getString("address");
					String storedType = results.getString("type");
					
					email.setAddress(storedAddress);
					
					if (storedType != null)
					{
						if (storedType.equals("ENRON1")){ email.setType(Type.ENRON1); }
						else if (storedType.equals("ENRON2")){ email.setType(Type.ENRON2); }
						else if (storedType.equals("ENRON3")){ email.setType(Type.ENRON3); }
						else if (storedType.equals("NOT_ENRON_PRO")){ email.setType(Type.NOT_ENRON_PRO); }
						else if (storedType.equals("PRIVATE")){ email.setType(Type.PRIVATE); }
					}
					
					emails.add(email);
				}
			}
			
			return emails;
		}
		catch(Exception e)
		{
			log.error("Unable to get actor emails: " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * get message TO, CC and BCC addresses
	 */
	public Address[] getMessageAddresses(String tableName, int messageKey)
	{
		try
		{
			Address[] addressArray = null;
			
			PreparedStatement s = this.conn.prepareStatement("SELECT address, type FROM email LEFT JOIN " + tableName + " ON email.key = " + tableName + ".email_key WHERE " + tableName + ".message_key = ?");
			s.setInt(1, messageKey);
			
			ResultSet results = this.execute(s);
			
			Vector<Address> addresses = new Vector<Address>();
			
			if (results != null)
			{
				while(results.next())
				{
					Address address = new InternetAddress(results.getString("address"));
					addresses.add(address);
				}
			}
			
			if (addresses.size() > 0)
			{
				int count = 0;
				addressArray = new Address[addresses.size()];
				
				for (Address address : addresses)
				{
					addressArray[count] = address;
					count++;
				}
			}
			
			return addressArray;
		}
		catch(Exception e)
		{
			log.error("Unable to get message emails: " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * get message
	 */
	public EnronMessage getMessageById(String id)
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT message.key, sender_key, sent_date, subject, content, relative_path, number_of_to_recipients,  number_of_cc_recipients,  number_of_bcc_recipients FROM message WHERE id = ?");
			s.setString(1, id);
			
			ResultSet results = this.execute(s);
			
			EnronMessage message = null;
			
			if (results != null && results.next())
			{
				int messageKey = results.getInt("key");
				int senderKey = results.getInt("sender_key");
				
				message = new EnronMessage();
				message.setId(id);
				message.setSentDate(results.getDate("sent_date"));
				message.setSubject(results.getString("subject"));
				message.setMimeBodyPart(results.getString("content"));
				message.setRelativeFilePath(results.getString("relative_path"));
				message.setNumberOfToRecipients(results.getInt("number_of_to_recipients"));
				message.setNumberOfCcRecipients(results.getInt("number_of_cc_recipients"));
				message.setNumberOfBccRecipients(results.getInt("number_of_bcc_recipients"));
				
				Address[] toAddresses = this.getMessageAddresses("message_to", messageKey);
				Address[] ccAddresses = this.getMessageAddresses("message_cc", messageKey);
				Address[] bccAddresses = this.getMessageAddresses("message_bcc", messageKey);
				
				message.setToRecipients(toAddresses);
				message.setCcRecipients(ccAddresses);
				message.setBccRecipients(bccAddresses);
				
				Email senderEmail = this.getEmailByKey(senderKey);
				message.setFrom(senderEmail.getAddress());
			}
			
			return message;
		}
		catch(Exception e)
		{
			log.error("Unable to get message by id ('" + id + "'): " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * get the list of all emails
	 */
	public Hashtable<Integer, Address> getAllEmails()
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT address FROM email");
			s.setFetchSize(DEFAULT_FETCH_SIZE);
			
			ResultSet results = this.execute(s);
			
			Hashtable<Integer, Address> emails = new Hashtable<Integer, Address>();
			
			if (results != null)
			{
				while (results.next())
				{
					String storedAddress = results.getString("address");
					int hash = storedAddress.hashCode();
					
					if (!emails.contains(hash))
					{
						Address address = new InternetAddress(storedAddress);
						emails.put(storedAddress.hashCode(), address);
					}
				}
			}
			
			return emails;
		}
		catch(Exception e)
		{
			log.error("Unable to get list of all emails: " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * get the list of all messages
	 * 
	 * NB: loading the whole set of messages is slow!
	 */
	@Deprecated
	public List<EnronMessage> getAllMessages()
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT message.key, id, sender_key, sent_date, subject, content, relative_path, number_of_to_recipients,  number_of_cc_recipients,  number_of_bcc_recipients FROM message");
			s.setFetchSize(DEFAULT_FETCH_SIZE);
			
			ResultSet results = this.execute(s);
			
			Vector<EnronMessage> messages = new Vector<EnronMessage>();
			
			if (results != null)
			{
				while (results.next())
				{
					int messageKey = results.getInt("key");
					int senderKey = results.getInt("sender_key");
					
					EnronMessage message = new EnronMessage();
					message.setId(results.getString("id"));
					message.setSentDate(results.getDate("sent_date"));
					message.setSubject(results.getString("subject"));
					message.setMimeBodyPart(results.getString("content"));
					message.setRelativeFilePath(results.getString("relative_path"));
					message.setNumberOfToRecipients(results.getInt("number_of_to_recipients"));
					message.setNumberOfCcRecipients(results.getInt("number_of_cc_recipients"));
					message.setNumberOfBccRecipients(results.getInt("number_of_bcc_recipients"));
					
					Address[] toAddresses = this.getMessageAddresses("message_to", messageKey);
					Address[] ccAddresses = this.getMessageAddresses("message_cc", messageKey);
					Address[] bccAddresses = this.getMessageAddresses("message_bcc", messageKey);
					
					message.setToRecipients(toAddresses);
					message.setCcRecipients(ccAddresses);
					message.setBccRecipients(bccAddresses);
					
					Email senderEmail = this.getEmailByKey(senderKey);
					message.setFrom(senderEmail.getAddress());
					
					messages.add(message);
				}
			}
			
			return messages;
		}
		catch(Exception e)
		{
			log.error("Unable to get list of all messages: " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * store the list of Enron links
	 */
	public boolean storeAllLinks()
	{
		try
		{
			// address -> actor hashtable
			Hashtable<String, Actor> actors = new Hashtable<String, Actor>();
			
			PreparedStatement s = this.conn.prepareStatement("SELECT message.key, id, sender_key, sent_date, subject, content, relative_path, number_of_to_recipients,  number_of_cc_recipients,  number_of_bcc_recipients FROM message");
			s.setFetchSize(DEFAULT_FETCH_SIZE);
			
			ResultSet results = this.execute(s);
			
			if (results != null)
			{
				while (results.next())
				{
					int messageKey = results.getInt("key");
					int senderKey = results.getInt("sender_key");
					
					EnronMessage message = new EnronMessage();
					
					message.setId(results.getString("id"));
					message.setSentDate(results.getDate("sent_date"));
					message.setSubject(results.getString("subject"));
					message.setMimeBodyPart(results.getString("content"));
					message.setRelativeFilePath(results.getString("relative_path"));
					message.setNumberOfToRecipients(results.getInt("number_of_to_recipients"));
					message.setNumberOfCcRecipients(results.getInt("number_of_cc_recipients"));
					message.setNumberOfBccRecipients(results.getInt("number_of_bcc_recipients"));
					
					Address[] toRecipients = this.getMessageAddresses("message_to", messageKey);
					Address[] ccRecipients = this.getMessageAddresses("message_cc", messageKey);
					Address[] bccRecipients = this.getMessageAddresses("message_bcc", messageKey);
					
					message.setToRecipients(toRecipients);
					message.setCcRecipients(ccRecipients);
					message.setBccRecipients(bccRecipients);
					
					Email senderEmail = this.getEmailByKey(senderKey);
					message.setFrom(senderEmail.getAddress());
					
					String fromAddress = senderEmail.getAddress().toString();
					
					// TO recipients
					if (toRecipients != null && toRecipients.length > 0)
					{
						for (int i = 0; i < toRecipients.length; i++)
						{
							String toAddress = toRecipients[i].toString();
							
							if (Email.isEnronEmail(fromAddress) && Email.isEnronEmail(toAddress))
							{
								Actor fromActor = actors.get(fromAddress);
								
								if (fromActor == null)
								{
									fromActor = this.getActorByAddress(fromAddress);
								}
								
								Actor toActor = actors.get(toAddress);
								
								if (toActor == null)
								{
									toActor = this.getActorByAddress(toAddress);
								}
								
								if (fromActor != null && toActor != null)
								{
									actors.put(fromAddress, fromActor);
									actors.put(toAddress, toActor);
									
									if (!fromActor.getId().equals(toActor.getId()))
									{
										log.info("Storing 'TO' link from actor '" + fromActor.getId() + "' to actor '" + toActor.getId() + "'");
										
										Link link = new Link();
										link.setFromActorId(fromActor.getId());
										
										// remove '<' & '>' characters from messageId
										// <messageId> -> messageId
										link.setLinkId(message.getId().replace("<", "").replace(">", ""));
										
										link.setStartDate(message.getSentDate());
										link.setType(Message.RecipientType.TO);
										link.setToActorId(toActor.getId());
										
										this.storeLink(link);
									}
								}
							}
						}
					} // TO
					
					// CC recipients
					if (ccRecipients != null && ccRecipients.length > 0)
					{
						for (int i = 0; i < ccRecipients.length; i++)
						{
							String toAddress = ccRecipients[i].toString();
							
							if (Email.isEnronEmail(fromAddress) && Email.isEnronEmail(toAddress))
							{
								Actor fromActor = actors.get(fromAddress);
								
								if (fromActor == null)
								{
									fromActor = this.getActorByAddress(fromAddress);
								}
								
								Actor toActor = actors.get(toAddress);
								
								if (toActor == null)
								{
									toActor = this.getActorByAddress(toAddress);
								}
								
								if (fromActor != null && toActor != null)
								{
									actors.put(fromAddress, fromActor);
									actors.put(toAddress, toActor);
									
									if (!fromActor.getId().equals(toActor.getId()))
									{
										log.info("Storing 'CC' link from actor '" + fromActor.getId() + "' to actor '" + toActor.getId() + "'");
										
										Link link = new Link();
										link.setFromActorId(fromActor.getId());
										
										// remove '<' & '>' characters from messageId
										// <messageId> -> messageId
										link.setLinkId(message.getId().replace("<", "").replace(">", ""));
										
										link.setStartDate(message.getSentDate());
										link.setType(Message.RecipientType.CC);
										link.setToActorId(toActor.getId());
										
										this.storeLink(link);
									}
								}
							}
						} 
					} // CC
					
					// BCC recipients
					if (bccRecipients != null && bccRecipients.length > 0)
					{
						for (int i = 0; i < bccRecipients.length; i++)
						{
							String toAddress = bccRecipients[i].toString();
							
							if (Email.isEnronEmail(fromAddress) && Email.isEnronEmail(toAddress))
							{
								Actor fromActor = actors.get(fromAddress);
								
								if (fromActor == null)
								{
									fromActor = this.getActorByAddress(fromAddress);
								}
								
								Actor toActor = actors.get(toAddress);
								
								if (toActor == null)
								{
									toActor = this.getActorByAddress(toAddress);
								}
								
								if (fromActor != null && toActor != null)
								{
									actors.put(fromAddress, fromActor);
									actors.put(toAddress, toActor);
									
									if (!fromActor.getId().equals(toActor.getId()))
									{
										log.info("Storing 'BCC' link from actor '" + fromActor.getId() + "' to actor '" + toActor.getId() + "'");
										
										Link link = new Link();
										link.setFromActorId(fromActor.getId());
										
										// remove '<' & '>' characters from messageId
										// <messageId> -> messageId
										link.setLinkId(message.getId().replace("<", "").replace(">", ""));
										
										link.setStartDate(message.getSentDate());
										link.setType(Message.RecipientType.BCC);
										link.setToActorId(toActor.getId());
										
										this.storeLink(link);
									}
								}
							}
						}
					} // BCC
				}
			}
			
			return true;
		}
		catch(Exception e)
		{
			log.error("Unable to store the set of all links: " + e.getMessage());
			e.printStackTrace();
			return false;
		}
	}
	
	/**
	 * get messages with a unique sender and a unique recipient 
	 */
	public List<EnronMessage> getUniqueRecipientMessages()
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT message.key, id, sender_key, sent_date, subject, content, relative_path, number_of_to_recipients,  number_of_cc_recipients,  number_of_bcc_recipients FROM message WHERE number_of_to_recipients = 1 AND number_of_cc_recipients = 0 AND number_of_bcc_recipients = 0");
			s.setFetchSize(DEFAULT_FETCH_SIZE);
			
			ResultSet results = this.execute(s);
			
			Vector<EnronMessage> messages = new Vector<EnronMessage>();
			
			if (results != null && results.next())
			{
				while (results.next())
				{
					int messageKey = results.getInt("key");
					int senderKey = results.getInt("sender_key");
					
					EnronMessage message = new EnronMessage();
					message.setId(results.getString("id"));
					message.setSentDate(results.getDate("sent_date"));
					message.setSubject(results.getString("subject"));
					message.setMimeBodyPart(results.getString("content"));
					message.setRelativeFilePath(results.getString("relative_path"));
					message.setNumberOfToRecipients(results.getInt("number_of_to_recipients"));
					message.setNumberOfCcRecipients(results.getInt("number_of_cc_recipients"));
					message.setNumberOfBccRecipients(results.getInt("number_of_bcc_recipients"));
					
					Address[] toAddresses = this.getMessageAddresses("message_to", messageKey);
					Address[] ccAddresses = this.getMessageAddresses("message_cc", messageKey);
					Address[] bccAddresses = this.getMessageAddresses("message_bcc", messageKey);
					
					message.setToRecipients(toAddresses);
					message.setCcRecipients(ccAddresses);
					message.setBccRecipients(bccAddresses);
					
					Email senderEmail = this.getEmailByKey(senderKey);
					message.setFrom(senderEmail.getAddress());
					
					messages.add(message);
				}
			}
			
			return messages;
		}
		catch(Exception e)
		{
			log.error("Unable to get list of unique recipient messages: " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * get messages with a unique sender and a unique recipient 
	 */
	public List<EnronMessage> getUniqueRecipientMessagesInDateRange(Date beginDate, Date endDate)
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT message.key, id, sender_key, sent_date, subject, content, relative_path, number_of_to_recipients,  number_of_cc_recipients,  number_of_bcc_recipients FROM message WHERE number_of_to_recipients = 1 AND number_of_cc_recipients = 0 AND number_of_bcc_recipients = 0 AND sent_date >= ? AND sent_date < ?");
			s.setDate(1, new java.sql.Date(beginDate.getTime()));
			s.setDate(2, new java.sql.Date(endDate.getTime()));
			s.setFetchSize(DEFAULT_FETCH_SIZE);
			
			ResultSet results = this.execute(s);
			
			Vector<EnronMessage> messages = new Vector<EnronMessage>();
			
			if (results != null && results.next())
			{
				while (results.next())
				{
					int messageKey = results.getInt("key");
					int senderKey = results.getInt("sender_key");
					
					EnronMessage message = new EnronMessage();
					message.setId(results.getString("id"));
					message.setSentDate(results.getDate("sent_date"));
					message.setSubject(results.getString("subject"));
					message.setMimeBodyPart(results.getString("content"));
					message.setRelativeFilePath(results.getString("relative_path"));
					message.setNumberOfToRecipients(results.getInt("number_of_to_recipients"));
					message.setNumberOfCcRecipients(results.getInt("number_of_cc_recipients"));
					message.setNumberOfBccRecipients(results.getInt("number_of_bcc_recipients"));
					
					Address[] toAddresses = this.getMessageAddresses("message_to", messageKey);
					Address[] ccAddresses = this.getMessageAddresses("message_cc", messageKey);
					Address[] bccAddresses = this.getMessageAddresses("message_bcc", messageKey);
					
					message.setToRecipients(toAddresses);
					message.setCcRecipients(ccAddresses);
					message.setBccRecipients(bccAddresses);
					
					Email senderEmail = this.getEmailByKey(senderKey);
					message.setFrom(senderEmail.getAddress());
					
					messages.add(message);
				}
			}
			
			return messages;
		}
		catch(Exception e)
		{
			log.error("Unable to get list of unique recipient messages in date range: " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * get Enron links in date range
	 */
	public List<Link> getLinksInDateRange(Date beginDate, Date endDate)
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT from_actor_id, message_id, time, type, to_actor_id FROM link WHERE time >= ? AND time < ?");
			s.setDate(1, new java.sql.Date(beginDate.getTime()));
			s.setDate(2, new java.sql.Date(endDate.getTime()));
			s.setFetchSize(DEFAULT_FETCH_SIZE);
			
			ResultSet results = this.execute(s);
			
			Vector<Link> links = new Vector<Link>();
			
			if (results != null && results.next())
			{
				while (results.next())
				{
					Link link = new Link();
					link.setFromActorId(results.getString("from_actor_id"));
					link.setLinkId(results.getString("message_id"));
					
					String type = results.getString("type");
					
					if (type.equals("TO")){ link.setType(RecipientType.TO); }
					else if (type.equals("CC")){ link.setType(RecipientType.CC); }
					else if (type.equals("BCC")){ link.setType(RecipientType.BCC); }
					
					link.setStartDate(results.getDate("time"));
					link.setToActorId(results.getString("to_actor_id"));
					
					links.add(link);
				}
			}
			
			return links;
		}
		catch(Exception e)
		{
			log.error("Unable to get list of Enron links in date range: " + e.getMessage());
			e.printStackTrace();
			return null;
		}
	}
	
	/**
	 * get the set of messages according to the specified from address
	 */
	public List<EnronMessage> getMessagesByFrom(String fromAddress)
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT message.key, id, sender_key, sent_date, subject, content, relative_path, number_of_to_recipients,  number_of_cc_recipients,  number_of_bcc_recipients FROM message LEFT JOIN email ON message.sender_key = email.key WHERE email.address = ?");
			s.setString(1, fromAddress);
			s.setFetchSize(DEFAULT_FETCH_SIZE);
			
			ResultSet results = this.execute(s);
			
			Vector<EnronMessage> messages = new Vector<EnronMessage>();
			
			if (results != null && results.next())
			{
				while (results.next())
				{
					int messageKey = results.getInt("key");
					int senderKey = results.getInt("sender_key");
					
					EnronMessage message = new EnronMessage();
					message.setId(results.getString("id"));
					message.setSentDate(results.getDate("sent_date"));
					message.setSubject(results.getString("subject"));
					message.setMimeBodyPart(results.getString("content"));
					message.setRelativeFilePath(results.getString("relative_path"));
					message.setNumberOfToRecipients(results.getInt("number_of_to_recipients"));
					message.setNumberOfCcRecipients(results.getInt("number_of_cc_recipients"));
					message.setNumberOfBccRecipients(results.getInt("number_of_bcc_recipients"));
					
					Address[] toAddresses = this.getMessageAddresses("message_to", messageKey);
					Address[] ccAddresses = this.getMessageAddresses("message_cc", messageKey);
					Address[] bccAddresses = this.getMessageAddresses("message_bcc", messageKey);
					
					message.setToRecipients(toAddresses);
					message.setCcRecipients(ccAddresses);
					message.setBccRecipients(bccAddresses);
					
					Email senderEmail = this.getEmailByKey(senderKey);
					message.setFrom(senderEmail.getAddress());
					
					messages.add(message);
				}
			}
			
			return messages;
		}
		catch(Exception e)
		{
			log.error("Unable to get list of messages sent by '" + fromAddress + "': " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * get the set of messages according to the specified to address
	 */
	public List<EnronMessage> getMessagesByTo(String toAddress)
	{
		try
		{
			PreparedStatement s = this.conn.prepareStatement("SELECT message.key, id, sender_key, sent_date, subject, content, relative_path, number_of_to_recipients,  number_of_cc_recipients,  number_of_bcc_recipients FROM message LEFT JOIN message_to ON message.key = message_to.message_key LEFT JOIN email ON message_to.email_key = email.key WHERE email.address = ?");
			s.setString(1, toAddress);
			s.setFetchSize(DEFAULT_FETCH_SIZE);
			
			ResultSet results = this.execute(s);
			
			Vector<EnronMessage> messages = new Vector<EnronMessage>();
			
			if (results != null && results.next())
			{
				while (results.next())
				{
					int messageKey = results.getInt("key");
					int senderKey = results.getInt("sender_key");
					
					EnronMessage message = new EnronMessage();
					message.setId(results.getString("id"));
					message.setSentDate(results.getDate("sent_date"));
					message.setSubject(results.getString("subject"));
					message.setMimeBodyPart(results.getString("content"));
					message.setRelativeFilePath(results.getString("relative_path"));
					message.setNumberOfToRecipients(results.getInt("number_of_to_recipients"));
					message.setNumberOfCcRecipients(results.getInt("number_of_cc_recipients"));
					message.setNumberOfBccRecipients(results.getInt("number_of_bcc_recipients"));
					
					Address[] toAddresses = this.getMessageAddresses("message_to", messageKey);
					Address[] ccAddresses = this.getMessageAddresses("message_cc", messageKey);
					Address[] bccAddresses = this.getMessageAddresses("message_bcc", messageKey);
					
					message.setToRecipients(toAddresses);
					message.setCcRecipients(ccAddresses);
					message.setBccRecipients(bccAddresses);
					
					Email senderEmail = this.getEmailByKey(senderKey);
					message.setFrom(senderEmail.getAddress());
					
					messages.add(message);
				}
			}
			
			return messages;
		}
		catch(Exception e)
		{
			log.error("Unable to get list of messages sent to '" + toAddress + "': " + e.getMessage());
			return null;
		}
	}
	
}
